package com.ry.flink.process;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.ry.flink.source.RedisSource;
import com.ry.flink.utils.Constants;
import com.ry.flink.utils.FlinkKafkaUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper;
import org.apache.flink.util.Collector;
import java.util.HashMap;
import java.util.Properties;

/**
 * 完成redis和kafka的数据拼接
 *  基于connect实现数据拼接
 * */
public class IndicatorProcessMain {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //根据Topic partition的数量进行设置
        env.setParallelism(1);
        //开启checkpoint机制，备份数据
        env.enableCheckpointing(20000);
        //设置仅一次语义
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //二次Checkpoint之间的间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
        //保留几个checkpoint的结果，保存1个就行了
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        //设置checkpint的超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        //程序失败或者完成是否保存checkpoint数据
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //设置state的保存策略
//        env.setStateBackend(new FsStateBackend());
        //设置重启策略
//        env.setRestartStrategy();


    }
}